package com.chenzhiling.fs.objectStorage;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.javaswift.joss.model.Account;
import org.javaswift.joss.model.Container;
import org.javaswift.joss.model.StoredObject;

import java.io.*;

/**
 * @Author: CHEN ZHI LING
 * @Date: 2021/10/15
 * @Description: ceph输出流
 */
public class CephObjectOutputStream extends OutputStream {

    private static final Log LOG = LogFactory.getLog(CephObjectOutputStream.class);
    //文件大小
    private final long filePartSize;
    private final Configuration conf;
    private final Account account;
    private final String key;
    //临时文件
    private File backupFile;
    //输出流
    private OutputStream backupStream;
    private boolean closed;
    //分区号
    private int partNumber;
    //block偏移量
    private long blockOffset;
    private long bytesWritten;
    //分区上传字节
    private long bytesUploaded;
    private boolean partUpload = false;
    final byte[] oneByte = new byte[1];

    public CephObjectOutputStream(Configuration conf, String key, long partSizeKB, Account account) throws IOException {
        this.conf = conf;
        this.key = key;
        this.account = account;
        this.backupFile = this.newBackupFile();
        this.backupStream = new BufferedOutputStream(new FileOutputStream(this.backupFile));
        this.partNumber = 1;
        this.blockOffset = 0L;
        this.filePartSize = 1024L * partSizeKB;
    }

    private File newBackupFile() throws IOException {
        //建立本地临时文件
        File dir = new File(this.conf.get("hadoop.tmp.dir"));
        if (!dir.mkdirs() && !dir.exists()) {
            throw new IOException("Cannot create Ceph buffer directory: " + dir);
        } else {
            File result = File.createTempFile("output-", ".tmp", dir);
            result.deleteOnExit();
            return result;
        }
    }

    public void flush() throws IOException {
        this.backupStream.flush();
    }

    private synchronized void verifyOpen() throws IOException {
        if (this.closed) {
            throw new IOException("stream is closed");
        }
    }

    public synchronized void close() throws IOException {
        if (!this.closed) {
            try {
                this.closed = true;
                this.backupStream.close();
                this.backupStream = null;
                Path keyPath = new Path(this.key);
                if (this.partUpload) {
                    this.partUpload(true);
                    //this.nativeStore.createManifestForPartUpload(keyPath);
                } else {
                    this.uploadOnClose(keyPath);
                }
            } finally {
                this.delete(this.backupFile);
                this.backupFile = null;
            }

            assert this.backupStream == null : "backup stream has been reopened";

        }
    }

    private void uploadOnClose(Path keyPath) throws IOException {
        boolean uploadSuccess = false;
        int attempt = 0;
        while(!uploadSuccess) {
            try {
                ++attempt;
                //上传文件
                this.bytesUploaded += this.uploadFileAttempt(keyPath);
                uploadSuccess = true;
            } catch (IOException var5) {
                LOG.info("Upload failed " + var5, var5);
                if (attempt > 3) {
                    throw var5;
                }
            }
        }

    }

    //上传文件
    private long uploadFileAttempt(Path keyPath) throws IOException {
        long uploadLen = this.backupFile.length();
        this.uploadFile(keyPath,new FileInputStream(this.backupFile));
        return uploadLen;
    }

    protected void finalize() {
        if (!this.closed) {
            LOG.warn("stream not closed");
        }

        if (this.backupFile != null) {
            LOG.warn("Leaking backing file " + this.backupFile);
        }

    }

    private void delete(File file) {
        if (file != null) {
            if (!file.delete()) {
                LOG.warn("Could not delete " + file);
            }
        }
    }


    public void write(int b) throws IOException {
        this.oneByte[0] = (byte)b;
        this.write(this.oneByte, 0, 1);
    }


    public synchronized void write(byte[] buffer, int offset, int len) throws IOException {
        if (offset >= 0 && len >= 0 && offset + len <= buffer.length) {
            this.verifyOpen();

            while(this.blockOffset + (long)len >= this.filePartSize) {
                int subWriteLen = (int)(this.filePartSize - this.blockOffset);
                if (subWriteLen < 0 || subWriteLen > len) {
                    throw new IOException("Invalid subWrite len: " + subWriteLen + " -buffer len: " + len);
                }
                this.writeToBackupStream(buffer, offset, subWriteLen);
                offset += subWriteLen;
                len -= subWriteLen;
                this.partUpload(false);
            }

            this.writeToBackupStream(buffer, offset, len);
        } else {
            throw new IndexOutOfBoundsException("Invalid offset/length for write");
        }
    }

    private void writeToBackupStream(byte[] buffer, int offset, int len) throws IOException {
        assert len >= 0 : "remainder to write is negative";

        if (len != 0) {
            this.backupStream.write(buffer, offset, len);
            this.blockOffset += len;
            this.bytesWritten += len;
        }
    }

    private void partUpload(boolean closingUpload) throws IOException {
        if (this.backupStream != null) {
            this.backupStream.close();
        }

        if (closingUpload && this.partUpload && this.backupFile.length() == 0L) {
            this.delete(this.backupFile);
        } else {
            this.partUpload = true;
            boolean uploadSuccess = false;
            int attempt = 0;

            while(!uploadSuccess) {
                try {
                    ++attempt;
                    this.bytesUploaded += this.uploadFilePartAttempt();
                    uploadSuccess = true;
                } catch (IOException var5) {
                    LOG.info("Upload failed " + var5, var5);
                    if (attempt > 3) {
                        throw var5;
                    }
                }
            }
            this.delete(this.backupFile);
            ++this.partNumber;
            this.blockOffset = 0L;
            if (!closingUpload) {
                this.backupFile = this.newBackupFile();
                this.backupStream = new BufferedOutputStream(new FileOutputStream(this.backupFile));
            }
        }

    }

    private long uploadFilePartAttempt() throws IOException {
        long uploadLen = this.backupFile.length();
        this.uploadFilePart(new Path(this.key), this.partNumber, new FileInputStream(this.backupFile));
        return uploadLen;
    }

    long getFilePartSize() {
        return this.filePartSize;
    }

    synchronized int getPartitionsWritten() {
        return this.partNumber - 1;
    }

    long getBytesWritten() {
        return this.bytesWritten;
    }

    long getBytesUploaded() {
        return this.bytesUploaded;
    }

    private void uploadFile(Path path, InputStream inputStream) throws IOException {
        CephObjectAttributes cephFiLe = CephObjectUtil.getCephFiLeByPath(path);
        String bucketName = cephFiLe.getBucketName();
        String objectName = cephFiLe.getObjectName();
        Container container = account.getContainer(bucketName);
        StoredObject object = container.getObject(objectName);
        object.uploadObject(inputStream);
    }

    private void uploadFilePart(Path path, int partNumber, InputStream inputStream) throws IOException {
        String stringPath = path.toUri().toString();
        String partitionFilename = this.partitionFilenameFromNumber(partNumber);
        if (stringPath.endsWith("/")) {
            stringPath = stringPath.concat(partitionFilename);
        } else {
            stringPath = stringPath.concat("/").concat(partitionFilename);
        }
        this.uploadFile(new Path(stringPath),inputStream);
    }

    private String partitionFilenameFromNumber(int partNumber) {
        return String.format("%06d", partNumber);
    }
}
